package com.sbp.message.service;

import com.alibaba.dubbo.common.utils.StringUtils;
import com.dap.dao.model.Page;
import com.dap.utils.IdGenerator;
import com.sbp.message.api.common.CommonUtils;
import com.sbp.message.api.common.TopicUtils;
import com.sbp.message.api.entity.*;
import com.sbp.message.api.service.MessageService;
import com.sbp.message.send_message.MessageSender;
import com.sbp.message.service.cs.MessageCSImpl;
import com.sbp.message.service.cs.TopicRegistryCSImpl;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.dao.DuplicateKeyException;
import org.springframework.stereotype.Service;

import java.util.*;

@Service("messageService")
public class MessageServiceImpl implements MessageService {
    private static final Logger logger = LoggerFactory.getLogger(MessageServiceImpl.class);

    @Autowired
    private TopicRegistryCSImpl topicRegistryCSImpl;
    @Autowired
    private MessageCSImpl messageCSImpl;

    @Autowired
    private MessageSender messageSender;
    @Autowired
    private ErrorNotifyServiceImpl errorNotifyServiceImpl;


    @Override
    public Message createAndSendTransparentMessage(Message message) {
        message.setIsPersistence(0);
        Date date = new Date();
        message.setCreateTime(date);
        message.setUpdateTime(date);
        message.setStatus(MessageStatus.MessageStatus_sending.getValue());
        messageSender.sendMessage(message);
        return message;
    }


    private Message createMessageWithStatus(Message message, MessageStatus messageStatus) {
        message.setIsPersistence(1);
        message.setId(String.valueOf(IdGenerator.generateId()));
        message.setStatus(messageStatus.getValue());

        // 如果是topic消息, 初始化的时候就保存需要消费的消费端
        if (message.getMessageType() == MessageType.MESSAGE_TOPIC.getValue()) {
            TopicRegistry topicRegistry = topicRegistryCSImpl.selectTopicListByDestination(message.getDestination());
            if (null != topicRegistry) {
                message.setConsumers(topicRegistry.getConsumers());
                message.setConsumersDescribe(topicRegistry.getConsumersDescribe());
            } else {
                message.setConsumers(0);
                message.setConsumersDescribe("");
            }
        }

        return messageCSImpl.insertMessage(message);
    }
    @Override
    public Message createMessage(Message message) {
        return createMessageWithStatus(message, MessageStatus.MessageStatus_unconfirmed);
    }
    // 创建并发送消息
    @Override
    public Message createAndSendMessage(Message message) {
        message = createMessageWithStatus(message, MessageStatus.MessageStatus_sending);
        messageSender.sendMessage(message);
        return message;
    }


    // 修改消息状态和错误日志
    @Override
    public int updateMessageStatus(String messageId, MessageStatus messageStatus, String errorMessage, String errorLog) {
        return messageCSImpl.updateMessageStatus(messageId, messageStatus, errorMessage, errorLog);
    }


    @Override
    public int tryEndMessageWithConsumeSucceed(Message message, String appName) {
        // queue, 直接删除
        if (message.getMessageType() == MessageType.MESSAGE_QUEUE.getValue()) {
            return messageCSImpl.deleteMessageById(message.getId());
        }

        // topic, 做以下操作
        else {
            // 合法消费者
            if (message.getConsumersDescribe().contains(appName)) {
                // 如果未消费, acknowledges++, acknowledges_describe 追加 appName
                int updateRowCount = messageCSImpl.updateMessageWithConsumeSucceedById(message.getId(), appName);
                if (updateRowCount <= 0)
                    return 0;

                // 如果 acknowledges >= consumers, 则将消息正常结束
                int deleteRowCount = messageCSImpl.deleteMessageByIdIfAllAcknowledged(message.getId());
                if (deleteRowCount > 0)
                    return deleteRowCount;

                // 尝试异常结束
                return messageCSImpl.endMessageByIdWithBusinessFailed(message.getId(), MessageStatus.MessageStatus_dead);
            }

            // 意外消费者
            else {
                messageCSImpl.addSucceedUnexpectedConsumer(message.getId(), appName);
                return 1;
            }
        }
    }
    @Override
    public int tryEndMessageWithBusinessFailed(Message message, String appName, String errorMessage, String errorLog) {
        try {
            // queue, kill消息
            if (message.getMessageType() == MessageType.MESSAGE_QUEUE.getValue()) {
                return messageCSImpl.updateMessageStatus(message.getId(), MessageStatus.MessageStatus_dead, errorMessage, errorLog);
            }

            // topic, 做以下操作
            else {
                // 合法消费者
                if (message.getConsumersDescribe().contains(appName)) {
                    int updateRowCount = messageCSImpl.updateMessageWithBusinessFailedById(message.getId(), appName, errorMessage, errorLog);
                    if (updateRowCount <= 0)
                        return 0;

                    // 尝试异常结束
                    return messageCSImpl.endMessageByIdWithBusinessFailed(message.getId(), MessageStatus.MessageStatus_dead);
                }

                // 意外消费者
                else {
                    messageCSImpl.addFailedUnexpectedConsumer(message.getId(), appName, errorLog);
                    return 1;
                }
            }
        } finally {
            errorNotifyServiceImpl.notify("业务异常", errorMessage, message.getId());
        }
    }

    @Override
    public int updateTopicRegistryByConsumersDescribe(String consumersDescribe) {
        String destination = TopicUtils.generateDestinationFromConsumersDescribe(consumersDescribe);

        TopicRegistry topic = topicRegistryCSImpl.selectTopicListByDestination(destination);
        if (topic != null) {
            if (StringUtils.isEmpty(topic.getConsumersDescribe()) || !topic.getConsumersDescribe().contains(consumersDescribe)) {
                return topicRegistryCSImpl.addConsumerById(topic.getId(), consumersDescribe);
            }
        } else {
            try {
                topic = new TopicRegistry();
                topic.setId(String.valueOf(IdGenerator.generateId()));
                topic.setConsumers(1);
                topic.setConsumersDescribe(consumersDescribe);
                topic.setDestination(destination);

                return topicRegistryCSImpl.insertTopicRegistry(topic);
            } catch (DuplicateKeyException e) {
                if (e.getMessage().indexOf("destination") >= 0)
                    return topicRegistryCSImpl.addConsumerById(topic.getId(), consumersDescribe);
                else
                    throw e;
            }
        }

        return 0;
    }


    @Override
    public Message queryMessage(String msgId) {
        return messageCSImpl.selectById(msgId);
    }

    @Override
    public boolean killMessage(String msgId) {
        Message msg = queryMessage(msgId);
        if (null == msg)
            return false;

        return 1 == messageCSImpl.updateMessageStatus(msgId, MessageStatus.MessageStatus_dead, null, null);
    }

    @Override
    public boolean closeMessage(String msgId) {
        return 1 == messageCSImpl.updateMessageStatus(msgId, MessageStatus.MessageStatus_end, null, null);
    }


    private static void removeBusinessFailedConsumer(Message msg) {
        if (msg.getBusinessFailed() > 0) {
            ArrayList<String> list = CommonUtils.splitByComma(msg.getAcknowledgesDescribe());
            StringBuffer sb = new StringBuffer();
            for (String acknowledgeDescribeInDB : list) {
                if (!CommonUtils.isBusinessFailedConsumer(acknowledgeDescribeInDB)) {
                    sb.append(',');
                    sb.append(acknowledgeDescribeInDB);
                }
            }

            msg.setAcknowledgesDescribe(sb.toString());
            msg.setBusinessFailed(0);
        }
    }
    private Message reliveMessageToStatus(String msgId, MessageStatus messageStatus) {
        Message msg = messageCSImpl.selectByIdAndStatus(msgId, MessageStatus.MessageStatus_dead);
        if (null == msg)
            return null;

        msg.status = messageStatus.getValue();
        msg.retry = CommonUtils.MAX_RETIES;   // 重试次数设为最大 标记为复活
        removeBusinessFailedConsumer(msg);
        if (1 != messageCSImpl.update(msg))
            return null;
        return msg;
    }
    @Override
    public boolean reliveMessage(String msgId) {
        return null != reliveMessageToStatus(msgId, MessageStatus.MessageStatus_confirmed);
    }
    @Override
    public boolean reliveAndSendMessage(String msgId) {
        Message msg = reliveMessageToStatus(msgId, MessageStatus.MessageStatus_sending);
        if (null == msg)
            return false;
        messageSender.sendMessage(msg);
        return true;
    }


    @Override
    public Message confirmAndSendMessage(Message message) {
        message.setStatus(MessageStatus.MessageStatus_sending.getValue());
        if (1 == messageCSImpl.updateUnconfirmedMessage(message)) {
            messageSender.sendMessage(message);
            return message;
        } else {
            return null;
        }
    }
    @Override
    public Message confirmAndSendMessage(String msgId) {
        Message msg = messageCSImpl.selectByIdAndStatus(msgId, MessageStatus.MessageStatus_unconfirmed);
        if (null == msg)
            return null;

        return confirmAndSendMessage(msg);
    }

    @Override
    public int confirmAndSendMessages(List<Message> messageList) {
        if (null == messageList || messageList.isEmpty())
            return 0;

        messageList = messageCSImpl.confirmMessages(messageList, MessageStatus.MessageStatus_sending);

        for (Message message : messageList) {
            messageSender.sendMessage(message);
        }

        return messageList.size();
    }
    @Override
    public int confirmAndSendMessagesById(List<String> messageIdList) {
        List<Message> messageList = messageCSImpl.selectByIdsAndStatus(messageIdList, MessageStatus.MessageStatus_unconfirmed);
        return confirmAndSendMessages(messageList);
    }


    @Override
    public Message sendMessage(String msgId) {
        Message message = queryMessage(msgId);
        if (null == message)
            return null;

        messageSender.sendMessage(message);
        return message;
    }
    @Override
    public Message sendMessage(Message msg) {
        messageSender.sendMessage(msg);
        return msg;
    }
    @Override
    public int sendMessages(List<Message> msgList) {
        int res = 0;

        for (Message msg : msgList) {
            messageSender.sendMessage(msg);
            ++res;
        }

        return res;
    }
    @Override
    public int sendMessagesById(List<String> msgIdList) {
        Map<String, Object> parameter = new HashMap<>();
        parameter.put("idList", msgIdList);

        List<Message> msgList = messageCSImpl.selectByIdsAndStatus(msgIdList, null);
        return sendMessages(msgList);
    }


    @Override
    public int removeMessages(List<String> msgIdList) {
        return messageCSImpl.removeMessages(msgIdList);
    }


    @Override
    public Page<Message> queryMessagePage(Map<String, Object> parameter, Integer pageIndex, Integer pageSize) {
        return messageCSImpl.queryMessageList(parameter, pageIndex, pageSize);
    }
    @Override
    public Page<MessageVO> queryMessageVOPage(Map<String, Object> parameter, Integer pageIndex, Integer pageSize) {
        Page<Message> messagePage =  messageCSImpl.queryMessageList(parameter, pageIndex, pageSize);
        List<MessageVO> data = new ArrayList<>();
        for (Message dbData : messagePage.getElements()) {
            data.add(MessageVO.fromMessage(dbData));
        }
        return new Page<>(messagePage.getTotalCount(), messagePage.getPageSize(), messagePage.getPageIndex(), data);
    }

    @Override
    public Page<Message> findUnconsumedMessage(Integer pageIndex, Integer pageSize) {
        return messageCSImpl.page(MessageStatus.MessageStatus_unconfirmed, pageIndex, pageSize);
    }

    @Override
    public List<String> findUnconsumedMessageId(Integer pageIndex, Integer pageSize) {
        return messageCSImpl.idPage(MessageStatus.MessageStatus_unconfirmed, pageIndex, pageSize);
    }

}
